package com.example.canal.event.canal;

import com.alibaba.fastjson.JSONObject;
import com.example.canal.common.bean.CanalBean;
import com.example.canal.event.config.DbChangeProcessor;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.context.ApplicationEvent;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;
import org.springframework.util.Assert;

import java.lang.reflect.Constructor;


/**
 * @author LiYuhang
 * @version 0.1
 * @application
 * @Date 2021/8/8 18:26
 */
@Component
@Slf4j
public class CanalEventConsumer implements ApplicationContextAware, InitializingBean {


    private ApplicationContext applicationContext;


    /**
     * 监听的队列名称为：canaltopic
     *
     * @param consumer
     * @param ack
     */
    @KafkaListener(topics = "canaltopic", groupId = "topic.canal")
    public void receive(ConsumerRecord<?, ?> consumer,Acknowledgment ack) {
        String value = (String) consumer.value();
        log.info("topic名称:{},key:{},分区位置:{},下标:{},value:{}",consumer.topic(),consumer.key(),consumer.partition(),consumer.offset(),value);
        //转换为javaBean
        CanalBean canalBean = JSONObject.parseObject(value,CanalBean.class);
        processData(canalBean);
        ack.acknowledge();
    }

    /**
     * 根据tableName发布不同的事件
     *
     * @param canalBean
     */
    private void processData(CanalBean canalBean) {
        //获取是否是DDL语句
        boolean isDdl = canalBean.isDdl();
        if (isDdl) {
            return;
        }
        String tableName = canalBean.getTable();
        if (StringUtils.isBlank(tableName)) {
            return;
        }
        //region  根据 表名，发布具体的事件

        Class<? extends ApplicationEvent> springEvent = DbChangeProcessor.getDbChangeDetailEventByTableName(tableName);

        if (springEvent == null) return;
        Object eventObject = null;
        try {
            Constructor<?> constructor = springEvent.getConstructor(Object.class);
            eventObject = constructor.newInstance(canalBean);
        } catch (Exception e) {
            log.error(" 通过反射 发布事件失败 ： tableName:{},msg:{}",tableName,e.getMessage(),e);
        }
        if (eventObject != null) {
            applicationContext.publishEvent(eventObject);
        }
        //endregion

    }


    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        this.applicationContext = applicationContext;
    }

    @Override
    public void afterPropertiesSet() throws Exception {
        Assert.notNull(applicationContext,"applicationContext is null");
    }
}
